package com.mjf.spark.day02

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 读取外部文件指定分区
 *    -默认分区规则
 *      math.min(分配给应用的CPU核数,2)
 *
 *    -指定分区
 *      >1.在textFile方法中，第二个参数minPartitions，表示最小分区数
 *        注意：是最小，不是实际分区个数
 *      >2.在实际计算分区个数的时候，会根据文件的总大小和最小分区数进行相除
 *        &如果余数为0
 *          那么最小分区数，就是最终实际的分区数
 *        &如果余数不为0
 *          那么最小分区数，要计算(double) bytesRemaining)/splitSize > SPLIT_SLOP
 *          -SPLIT_SLOP:1.1
 *          -bytesRemaining:文件大小
 *          -splitSize:Math.max(minSize, Math.min(goalSize, blockSize))
 *              -minSize:默认1L
 *              -goalSize:文件总大小/分区数   totalSize / (numSplits == 0 ? 1 : numSplits)
 *
 */
object Spark05_Partition_file {
  def main(args: Array[String]): Unit = {

    // 创建SparkConf配置文件
    val conf = new SparkConf().setMaster("local[*]").setAppName("Spark05_Partition_file")
    // 创建SparkContext对象
    val sc = new SparkContext(conf)

    // 从本地文件中读取数据，创建RDD
    // 输入数据1 2 3 4，采用默认的分区方式，最终分区数2
    // 0号分区->1,2  1号分区->3,4
//    val rdd: RDD[String] = sc.textFile("D:\\code\\spark\\input\\2.txt")

    // 输入数据1 2 3 4，minPartitions设置为3，最终分区数4
    // 0号分区->1,2  1号分区->3  2号分区->4  3号分区->空
//    val rdd: RDD[String] = sc.textFile("D:\\code\\spark\\input\\2.txt", 3)
    val rdd: RDD[String] = sc.textFile("D:\\code\\spark\\input\\test1.txt", 5)
//    val rdd: RDD[String] = sc.textFile("D:\\code\\spark\\input\\test2.txt", 3)

    rdd.saveAsTextFile("D:\\code\\spark\\output")

    // 关闭连接
    sc. stop()

  }
}
